package com.xiaoniu.websocket;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.Vertx;
import io.vertx.core.http.HttpServer;
import io.vertx.core.http.HttpServerOptions;
import io.vertx.core.http.ServerWebSocket;
import io.vertx.core.http.WebSocketFrame;
import io.vertx.core.http.impl.ws.WebSocketFrameImpl;
import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;
import io.vertx.core.streams.ReadStream;

import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.stream.Stream;

/**
 * Created by sweet on 2018/3/8.
 * ---------------------------
 */
public class WebSocketServerVerticle extends AbstractVerticle {

  public static void main(String[] args) {
    System.setProperty("vertx.logger-delegate-factory-class-name",
            "io.vertx.core.logging.Log4j2LogDelegateFactory");
    Vertx.vertx().deployVerticle(new WebSocketServerVerticle());
  }

  private final Logger logger = LoggerFactory.getLogger(WebSocketServerVerticle.class);

  private final Map<String, ServerWebSocket> webSocketMap = new HashMap<>();

  // 用户ID : SOCKET ID
  private final Map<String, String> socketAndUserIdMap = new HashMap<>();

  @Override
  public void start() throws Exception {

    HttpServer httpServer = vertx.createHttpServer(new HttpServerOptions().setTcpKeepAlive(true));

    ReadStream<ServerWebSocket> stream = httpServer.websocketStream();

    stream.handler(socket -> {

      socket.closeHandler(r -> {
        logger.debug(socket.textHandlerID() + " --- 关闭");
        webSocketMap.remove(socket.textHandlerID());
        removeSocketAndUserIdMap(socket.textHandlerID());
      });

      String socketId = socket.textHandlerID();
      String path = socket.path();
      String query = socket.query();

      String[] split = query.split("=");
//      if (split.length != 2 || split[1].length() != 32) {
//        socket.close();
//        return;
//      }

      if (path.equals("/connect")) {
        putValue(socketId, socket, split[1]);
        socket.writeFrame(msgFrame("id: " + socketId));
      } else {
        socket.close();
      }

      socket.handler(buffer -> {
        String msg = buffer.toString();
        logger.debug(msg);
      });
    });

    stream.exceptionHandler(ex -> {
      ex.printStackTrace();
      logger.error("webSocket 服务发生错误，" + ex.getMessage());
    });
    stream.endHandler(end -> {
      logger.debug("socket end");
    });

    vertx.setPeriodic(1000l * 20, t -> {
      System.out.println("--------webSocketMap----------");
      webSocketMap.forEach((k,v) -> {
        System.out.println(k + " --- " + v);
        v.writeFrame(msgFrame("I'm Server " + LocalDateTime.now()));
      });
      System.out.println("--------------userIdMap-------------");
      socketAndUserIdMap.forEach((k,v) -> {
        System.out.println(k + " --- " + v);
      });
    });

    httpServer.listen(8080, server -> {
      if (server.succeeded()) {
        logger.info("webSocket 服务已经启动, port: " + 8080);
      } else {
        server.cause().printStackTrace();
      }
    });


  }

  private void removeSocketAndUserIdMap(String socketId) {
    socketAndUserIdMap.entrySet().removeIf(next -> socketId.equals(next.getValue()));
  }

  private void putValue(String socketId, ServerWebSocket socket, String s) {
    if (socket != null) {
      webSocketMap.put(socketId, socket);
    }
    if (s != null) {
      socketAndUserIdMap.put(s, socketId);
    }
  }

  private WebSocketFrame msgFrame(String msg) {
    return new WebSocketFrameImpl(msg, true);
  }
}
